package com.bw.gmall.realtime.dim.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.common.base.BaseApp;
import com.bw.gmall.realtime.common.bean.TableProcessDim;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.util.FlinkSourceUtil;
import com.bw.gmall.realtime.common.util.HbaseUtil;
import com.bw.gmall.realtime.dim.function.DimProcessFunction;
import com.bw.gmall.realtime.dim.function.DimSinkFunction;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.flink.api.java.tuple.Tuple2;


import java.io.IOException;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;

public class DimApp extends BaseApp {

    public static void main(String[] args) {
        new DimApp().start(Constant.TOPIC_DB,Constant.DIM_APP,4,10001);
    }





    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> dataStreamSource) {
        //1、ETL数据(下游数据需要啥,你过滤啥)
        SingleOutputStreamOperator<JSONObject> etlStream = getEtlStream(dataStreamSource);

        //2、通过flinkCDC读取mysql配置表，去掉Hbase建表,生成广播流
        DataStreamSource<String> processStream = env.fromSource(FlinkSourceUtil.getMysqlSource(Constant.PROCESS_DATABASE, Constant.PROCESS_DIM_TABLE_NAME), WatermarkStrategy.noWatermarks(), "cdc_stream").setParallelism(1);
        SingleOutputStreamOperator<TableProcessDim> createTableStream = createTable(processStream);
        MapStateDescriptor<String, TableProcessDim> mapDescriptor = new MapStateDescriptor<>("broadcast_state", String.class, TableProcessDim.class);
        BroadcastStream<TableProcessDim> broadcastStream  = createTableStream.broadcast(mapDescriptor);

        //3、用主流和广播流进行connect操作

        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcessDim>> connectstream = etlStream.connect(broadcastStream).process(new DimProcessFunction(mapDescriptor));

        //4、过滤字段(不是所有字段都写到Hbase里面)
        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcessDim>> filterStream = getFilterStream(connectstream);


        //5、写入Hbase

        filterStream.addSink(new DimSinkFunction());

    }


    private SingleOutputStreamOperator<Tuple2<JSONObject, TableProcessDim>> getFilterStream(SingleOutputStreamOperator<Tuple2<JSONObject, TableProcessDim>> processBroadCastStream) {
        return processBroadCastStream.map(new MapFunction<Tuple2<JSONObject, TableProcessDim>, Tuple2<JSONObject, TableProcessDim>>() {
            @Override
            public Tuple2<JSONObject, TableProcessDim> map(Tuple2<JSONObject, TableProcessDim> processDimTuple2) throws Exception {
                // 主流维度数据
                JSONObject f0 = processDimTuple2.f0;
                // 主流维度对于配置表
                TableProcessDim f1 = processDimTuple2.f1;
                // 要写入的列
                List<String> columns = Arrays.asList(f1.getSinkColumns().split(","));
                // 取出主流data
                JSONObject data = f0.getJSONObject("data");
                data.keySet().removeIf(key -> !columns.contains(key));
                return processDimTuple2;
            }
        });
    }


    public SingleOutputStreamOperator<TableProcessDim> createTable( DataStreamSource<String> processStream){


        return processStream.flatMap(new RichFlatMapFunction<String, TableProcessDim>() {

            private Connection hbaseConnect;
            private TableProcessDim tableProcessDim;


            // 用来链接数据库
            @Override
            public void open(Configuration parameters) throws Exception {
                // 连接Hbase -->ctrl+alt+f
                hbaseConnect = HbaseUtil.getHbaseConnect();
            }

            @Override
            public void close() throws Exception {
                // 关闭连接
                hbaseConnect.close();
            }

            /**
             *
             * @param s
             * @param collector
             * @throws Exception
             *
             * r 读
             * c 创建
             * d 删除
             * u 更新
             *
             * {"before":null,"after":{"source_table":"financial_sku_cost","sink_table":"dim_financial_sku_cost","sink_family":"info","sink_columns":"id,sku_id,sku_name,busi_date,is_lastest,sku_cost,create_time","sink_row_key":"id"},"source":{"version":"1.9.7.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":0,"snapshot":"false","db":"gmall2023_config","sequence":null,"table":"table_process_dim","server_id":0,"gtid":null,"file":"","pos":0,"row":0,"thread":null,"query":null},"op":"r","ts_ms":1740187564168,"transaction":null}
             */
            @Override
            public void flatMap(String s, Collector<TableProcessDim> collector) throws Exception {
                // 处理逻辑
                JSONObject jsonObject = JSON.parseObject(s);
                String op = jsonObject.getString("op");
                if ("r".equals(op) || "c".equals(op)) {
                    tableProcessDim = jsonObject.getObject("after", TableProcessDim.class);
                    String[] split = tableProcessDim.getSinkFamily().split(",");
                    // 创建表
                    createTable(split);
                } else if ("d".equals(op)) {
                    tableProcessDim = jsonObject.getObject("before", TableProcessDim.class);
                    // 删除表
                    deleteTable();
                } else if ("u".equals(op)) {
                    tableProcessDim = jsonObject.getObject("after", TableProcessDim.class);
                    String[] split = tableProcessDim.getSinkFamily().split(",");
                    // 先删除后建
                    deleteTable();
                    createTable(split);
                }
                tableProcessDim.setOp(op);
                collector.collect(tableProcessDim);
            }

            public void createTable(String[] families) {
                try {
                    System.out.println("tableProcessDim.getSinkTable() = " + tableProcessDim.getSinkTable());
                    HbaseUtil.createHBaseTable(hbaseConnect, Constant.HBASE_NAMESPACE, tableProcessDim.getSinkTable(), families);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }

            public void deleteTable() {
                System.out.println("tableProcessDim.getSinkTable() = " + tableProcessDim.getSinkTable());
                try {
                    HbaseUtil.dropHBaseTable(hbaseConnect, Constant.HBASE_NAMESPACE, tableProcessDim.getSinkTable());
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }).setParallelism(1);
    }


    private static SingleOutputStreamOperator<JSONObject> getEtlStream(DataStreamSource<String> dataStreamSource) {
        return dataStreamSource.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String s, Collector<JSONObject> collector) throws Exception {

                try {

                    if (s != null) {

                        JSONObject jsonObject = JSONObject.parseObject(s);
                        String database = jsonObject.getString("database");
                        String type = jsonObject.getString("type");
                        String data = jsonObject.getString("data");
                        if ("gmall".equals(database) && !"bootstrap-start".equals(type) && !"bootstrap-complete".equals(type) && data != null && data.length() > 0) {
                            collector.collect(jsonObject);
                        }
                    }

                } catch (Exception e) {
                    e.printStackTrace();
                }

            }
        });
    }
}
